package com.example.reactor.reactor;

import java.util.*;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Predicate;

public class Flux<T> implements Publisher<T> {
    private final List<T> values;

    public Flux(List<T> values) {
        this.values = values;
    }

    // map 操作符
    public <R> Flux<R> map(Function<? super T, ? extends R> mapper) {
        List<R> mappedValues = new ArrayList<>();
        for (T value : values) {
            mappedValues.add(mapper.apply(value));
        }
        return new Flux<>(mappedValues);
    }

    // flatMap 操作符
    public <R> Flux<R> flatMap(Function<? super T, Flux<R>> mapper) {
        List<R> resultList = new ArrayList<>();
        for (T value : values) {
            Flux<R> newFlux = mapper.apply(value);
            newFlux.subscribe(new MySubscriber<>(resultList::add, () -> {
            }, e -> {
            }));
        }
        return new Flux<>(resultList);
    }

    // filter 操作符
    public Flux<T> filter(Predicate<? super T> predicate) {
        List<T> filtered = new ArrayList<>();
        for (T value : values) {
            if (predicate.test(value)) {
                filtered.add(value);
            }
        }
        return new Flux<>(filtered);
    }

    // merge 操作符
    public static <T> Flux<T> merge(Flux<T> flux1, Flux<T> flux2) {
        List<T> merged = new ArrayList<>();
        flux1.subscribe(new MySubscriber<>(merged::add, () -> {
        }, e -> {
        }));
        flux2.subscribe(new MySubscriber<>(merged::add, () -> {
        }, e -> {
        }));
        return new Flux<>(merged);
    }

    // concat 操作符
    public static <T> Flux<T> concat(Flux<T> flux1, Flux<T> flux2) {
        List<T> concatenated = new ArrayList<>();
        flux1.subscribe(new MySubscriber<>(concatenated::add, () -> {
        }, e -> {
        }));
        flux2.subscribe(new MySubscriber<>(concatenated::add, () -> {
        }, e -> {
        }));
        return new Flux<>(concatenated);
    }

    // zip 操作符
    public static <T, U> Flux<Map.Entry<T, U>> zip(Flux<T> flux1, Flux<U> flux2) {
        List<Map.Entry<T, U>> zipped = new ArrayList<>();
        Iterator<T> iter1 = flux1.values.iterator();
        Iterator<U> iter2 = flux2.values.iterator();
        while (iter1.hasNext() && iter2.hasNext()) {
            zipped.add(new AbstractMap.SimpleEntry<>(iter1.next(), iter2.next()));
        }
        return new Flux<>(zipped);
    }

    // buffer 操作符
    public Flux<List<T>> buffer(int size) {
        List<List<T>> buffered = new ArrayList<>();
        List<T> currentBuffer = new ArrayList<>();
        for (T value : values) {
            currentBuffer.add(value);
            if (currentBuffer.size() == size) {
                buffered.add(new ArrayList<>(currentBuffer));
                currentBuffer.clear();
            }
        }
        if (!currentBuffer.isEmpty()) {
            buffered.add(currentBuffer);
        }
        return new Flux<>(buffered);
    }

    // doOnNext 操作符
    public Flux<T> doOnNext(Consumer<? super T> consumer) {
        for (T value : values) {
            consumer.accept(value);
        }
        return this;
    }

    // doOnComplete 操作符
    public Flux<T> doOnComplete(Runnable runnable) {
        runnable.run();
        return this;
    }

    // doOnError 操作符
    public Flux<T> doOnError(Consumer<Throwable> consumer) {
        // No error in Flux by default, so we just call the consumer with a null Throwable.
        consumer.accept(null);
        return this;
    }

    @Override
    public void subscribe(Subscriber<? super T> subscriber) {
        for (T value : values) {
            subscriber.onNext(value);
        }
        subscriber.onComplete();
    }
}